Glueを使ってRDSからPinpointのセグメント情報を抽出してみた

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrameCollection
from awsglue.dynamicframe import DynamicFrame
from awsglue import DynamicFrame
import re

# Script generated for node Custom transform
def add_colum2(glueContext, dfc) -> DynamicFrameCollection:
    from pyspark.sql.functions import col, concat, lit, regexp_replace

    df = dfc.select(list(dfc.keys())[0]).toDF()
    df = df.withColumn("ChannelType", lit("SMS"))
    df = df.withColumn("Address", regexp_replace("Address", "-", ""))
    df = df.withColumn("Address", regexp_replace("Address", "^.", ""))
    df_add_colum = df.withColumn("Address", concat(lit("+81"), col("Address")))
    output_df1 = DynamicFrame.fromDF(df_add_colum, glueContext, "output_df")
    return DynamicFrameCollection({"output_df1": output_df1}, glueContext)


# Script generated for node Custom transform
def MyTransform(glueContext, dfc) -> DynamicFrameCollection:
    selected = dfc.select(list(dfc.keys())[0]).toDF()
    reprep = selected.coalesce(1)
    results = DynamicFrame.fromDF(reprep, glueContext, "results")
    return DynamicFrameCollection({"results": results}, glueContext)


def sparkUnion(glueContext, unionType, mapping, transformation_ctx) -> DynamicFrame:
    for alias, frame in mapping.items():
        frame.toDF().createOrReplaceTempView(alias)
    result = spark.sql(
        "(select * from source1) UNION " + unionType + " (select * from source2)"
    )
    return DynamicFrame.fromDF(result, glueContext, transformation_ctx)


# Script generated for node Custom transform
def add_colum1(glueContext, dfc) -> DynamicFrameCollection:
    from pyspark.sql.functions import lit

    df = dfc.select(list(dfc.keys())[0]).toDF()
    df_add_colum = df.withColumn("ChannelType", lit("EMAIL"))
    output_df1 = DynamicFrame.fromDF(df_add_colum, glueContext, "output_df")
    return DynamicFrameCollection({"output_df1": output_df1}, glueContext)


args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# Script generated for node SQL Server table
SQLServertable_node1 = glueContext.create_dynamic_frame.from_catalog(
    database="mssql-s3-integration-dev-db",
    table_name="test_dbo_users",
    transformation_ctx="SQLServertable_node1",
)

# Script generated for node ApplyMapping
ApplyMapping_node2 = ApplyMapping.apply(
    frame=SQLServertable_node1,
    mappings=[("telephone", "string", "Address", "string")],
    transformation_ctx="ApplyMapping_node2",
)

# Script generated for node Apply Mapping
ApplyMapping_node1658129522250 = ApplyMapping.apply(
    frame=SQLServertable_node1,
    mappings=[("email", "string", "Address", "string")],
    transformation_ctx="ApplyMapping_node1658129522250",
)

# Script generated for node Filter
Filter_node1658143711633 = Filter.apply(
    frame=ApplyMapping_node2,
    f=lambda row: (bool(re.match("\S", row["Address"]))),
    transformation_ctx="Filter_node1658143711633",
)

# Script generated for node Filter
Filter_node1658143945724 = Filter.apply(
    frame=ApplyMapping_node1658129522250,
    f=lambda row: (bool(re.match("\S", row["Address"]))),
    transformation_ctx="Filter_node1658143945724",
)

# Script generated for node Custom transform
Customtransform_node1658140204230 = add_colum2(
    glueContext,
    DynamicFrameCollection(
        {"Filter_node1658143711633": Filter_node1658143711633}, glueContext
    ),
)

# Script generated for node Custom transform
Customtransform_node1658140211475 = add_colum1(
    glueContext,
    DynamicFrameCollection(
        {"Filter_node1658143945724": Filter_node1658143945724}, glueContext
    ),
)

# Script generated for node Select From Collection
SelectFromCollection_node1658140219909 = SelectFromCollection.apply(
    dfc=Customtransform_node1658140204230,
    key=list(Customtransform_node1658140204230.keys())[0],
    transformation_ctx="SelectFromCollection_node1658140219909",
)

# Script generated for node Select From Collection
SelectFromCollection_node1658140226309 = SelectFromCollection.apply(
    dfc=Customtransform_node1658140211475,
    key=list(Customtransform_node1658140211475.keys())[0],
    transformation_ctx="SelectFromCollection_node1658140226309",
)

# Script generated for node Union
Union_node1658134250028 = sparkUnion(
    glueContext,
    unionType="ALL",
    mapping={
        "source1": SelectFromCollection_node1658140226309,
        "source2": SelectFromCollection_node1658140219909,
    },
    transformation_ctx="Union_node1658134250028",
)

# Script generated for node Custom transform
Customtransform_node1658311524969 = MyTransform(
    glueContext,
    DynamicFrameCollection(
        {"Union_node1658134250028": Union_node1658134250028}, glueContext
    ),
)

# Script generated for node Select From Collection
SelectFromCollection_node1658311587979 = SelectFromCollection.apply(
    dfc=Customtransform_node1658311524969,
    key=list(Customtransform_node1658311524969.keys())[0],
    transformation_ctx="SelectFromCollection_node1658311587979",
)

# Script generated for node Amazon S3
AmazonS3_node1658134594283 = glueContext.write_dynamic_frame.from_options(
    frame=SelectFromCollection_node1658311587979,
    connection_type="s3",
    format="csv",
    connection_options={
        "path": "s3://mssql-s3-integration-dev-mssqls3integrationdevs30-q462kdal8dd1",
        "partitionKeys": [],
    },
    transformation_ctx="AmazonS3_node1658134594283",
)

job.commit()